// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/common/base/spill_config.h>
#include <pollux/common/base/exceptions.h>

namespace kumo::pollux::common {
    SpillConfig::SpillConfig(
        GetSpillDirectoryPathCB _getSpillDirPathCb,
        UpdateAndCheckSpillLimitCB _updateAndCheckSpillLimitCb,
        std::string _fileNamePrefix,
        uint64_t _maxFileSize,
        uint64_t _writeBufferSize,
        uint64_t _readBufferSize,
        melon::Executor *_executor,
        int32_t _minSpillableReservationPct,
        int32_t _spillableReservationGrowthPct,
        uint8_t _startPartitionBit,
        uint8_t _numPartitionBits,
        int32_t _maxSpillLevel,
        uint64_t _maxSpillRunRows,
        uint64_t _writerFlushThresholdSize,
        const std::string &_compressionKind,
        std::optional<PrefixSortConfig> _prefixSortConfig,
        const std::string &_fileCreateConfig)
        : getSpillDirPathCb(std::move(_getSpillDirPathCb)),
          updateAndCheckSpillLimitCb(std::move(_updateAndCheckSpillLimitCb)),
          fileNamePrefix(std::move(_fileNamePrefix)),
          maxFileSize(
              _maxFileSize == 0
                  ? std::numeric_limits<int64_t>::max()
                  : _maxFileSize),
          writeBufferSize(_writeBufferSize),
          readBufferSize(_readBufferSize),
          executor(_executor),
          minSpillableReservationPct(_minSpillableReservationPct),
          spillableReservationGrowthPct(_spillableReservationGrowthPct),
          startPartitionBit(_startPartitionBit),
          numPartitionBits(_numPartitionBits),
          maxSpillLevel(_maxSpillLevel),
          maxSpillRunRows(_maxSpillRunRows),
          writerFlushThresholdSize(_writerFlushThresholdSize),
          compressionKind(common::stringToCompressionKind(_compressionKind)),
          prefixSortConfig(_prefixSortConfig),
          fileCreateConfig(_fileCreateConfig) {
        POLLUX_USER_CHECK_GE(
            spillableReservationGrowthPct,
            minSpillableReservationPct,
            "Spillable memory reservation growth pct should not be lower than minimum available pct");
    }

    int32_t SpillConfig::spillLevel(uint8_t startBitOffset) const {
        POLLUX_CHECK_LE(
            startBitOffset + numPartitionBits,
            64,
            "startBitOffset:{} numPartitionsBits:{}",
            startBitOffset,
            numPartitionBits);
        const int32_t deltaBits = startBitOffset - startPartitionBit;
        POLLUX_CHECK_GE(deltaBits, 0, "deltaBits:{}", deltaBits);
        POLLUX_CHECK_EQ(
            deltaBits % numPartitionBits,
            0,
            "deltaBits:{} numPartitionsBits{}",
            deltaBits,
            numPartitionBits);
        return deltaBits / numPartitionBits;
    }

    bool SpillConfig::exceedSpillLevelLimit(uint8_t startBitOffset) const {
        if (startBitOffset + numPartitionBits > 64) {
            return true;
        }
        if (maxSpillLevel == -1) {
            return false;
        }
        return spillLevel(startBitOffset) > maxSpillLevel;
    }
} // namespace kumo::pollux::common
